/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.io;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import stanford.tagger.Dictionary;
import stanford.tagger.TagCount;

/**
 *
 * @author hadoop
 */
public class DictionaryReader implements HDFSReader {

    private String inputDir;
    private Dictionary dict;

    public DictionaryReader(String inputDir) {
        this.inputDir = inputDir;
        dict = new Dictionary();
    }

    public DictionaryReader() {
        dict = new Dictionary();
    }

    public String getInputDir() {
        return inputDir;
    }

    public void setInputDir(String inputDir) {
        this.inputDir = inputDir;
    }

    public Dictionary getDict() {
        return dict;
    }


    public void read() {
        try {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(conf);
            FileStatus[] inputFiles = fs.listStatus(new Path(inputDir));

            for (int i = 0; i < inputFiles.length; i++) {
                // don't process the log files
                if (!inputFiles[i].getPath().getName().equalsIgnoreCase("_logs")) {
                    // input stream
                    FSDataInputStream in = fs.open(inputFiles[i].getPath());

                   String lines;
                   String word;
                   TagCount tagCount = new TagCount();

                    while((lines=in.readLine()) != null) {
                        try{

                            word = lines.split("\t")[0];
                            tagCount.setMap(getTagCount(lines.split("\t")[1]));
                            dict.add(word,tagCount);

                        }catch(Exception e){
                            e.getMessage();
                        }
                    }
                    in.close();
                }

            }
        } catch (IOException iOException) {
            iOException.printStackTrace();
        }
    }

    public void print() {

        Iterator<String> dicIt = dict.getDict().keySet().iterator();              

        while(dicIt.hasNext()){
            String word = dicIt.next();
            System.out.print(word+": ");            

            String[] tags = dict.getTags(word);

            for(String tag:tags){
                System.out.print(tag+"=>"+dict.getCount(word, tag)+" ");
            }

            System.out.println();
        }
    }

    private HashMap<String, Integer> getTagCount(String line) {

        String[] tagstring = line.split(" ");
        String tag;
        Integer count;
        HashMap<String, Integer> tagCount = new HashMap<String, Integer>();

        for(String tagtoken: tagstring){
            tag = tagtoken.split("=>")[0];
            count = Integer.parseInt(tagtoken.split("=>")[1]);
            tagCount.put(tag, count);
        }
        
        return tagCount;
    }
    
    
    

}
